package com.cn.file;

import ch.qos.logback.core.net.SyslogOutputStream;
import com.cn.exception.MyException;

import java.io.File;
import java.nio.ByteBuffer;
import java.util.List;

/**
 * 消息队列索引存储
 */
public class ConsumeQueue implements DispatcherService {

    public static final int UNIT_SIZE = 20;

    private final DefaultMessageStore defaultMessageStore;

    private final MapedFileQueue mapedFileQueue;

    /**
     * 文件大小
     */
    private final int fileSize;

    //消息主题
    private final String topic;

    //消息队列id
    private final int queueId;

    //存储路径
    private final String storePath;

    //内存
    private final ByteBuffer byteBuffer;

    //当前文件中最后一条消息在commitlog中的物理位置
    private long maxPhysicOffset = -1;
    private volatile long minLogicOffset = 0;

    public ConsumeQueue(DefaultMessageStore defaultMessageStore, int fileSize, String topic, int queueId, String storePath) {
        this.defaultMessageStore = defaultMessageStore;
        this.fileSize = fileSize;
        this.topic = topic;
        this.queueId = queueId;
        this.storePath = storePath + File.separator + topic + File.separator + queueId + File.separator;
        this.byteBuffer = ByteBuffer.allocate(UNIT_SIZE);
        //初始化队列文件
        this.mapedFileQueue = new MapedFileQueue(this.storePath, this.defaultMessageStore.getMessageStoreConfig().getConsumeFileSize());
    }

    public boolean load() {
        this.mapedFileQueue.load();
        return true;
    }

    /**
     * 恢复consumequeue文件的写入位移
     */
    public void recover() {
        //逻辑文件列表
        List<MappFile> mappedFiles = this.mapedFileQueue.getMappedFiles();
        if (!mappedFiles.isEmpty()) {
            //逻辑文件个数
            int size = mappedFiles.size();
            //consumequeue文件可能存在多个，恢复位移的时候，不需要扫描全部的文件，只需要从倒数第二个恢复
            int index = size - 3;
            index = index <= 0 ? 0 : index;

            //consumeQueue文件的大小
            int mappFileSizeLogics = this.fileSize;
            MappFile mappFile = mappedFiles.get(index);
            ByteBuffer byteBuffer = mappFile.sliceByteBuffer();
            //逻辑文件起始位移
            int processOffset = (int) mappFile.getFileFromOffset();
            //queue文件可写入位置，queue文件中每条消息的大小是固定的20字节
            long mappedFileOffset = 0;

            while (true) {
                /**
                 * queue文件中每条记录大小都是固定的20字节，因此每迭代一次，算一条记录（每遍历一次就是一条记录）
                 */
                for (int i = 0; i < mappFileSizeLogics; i += UNIT_SIZE) {
                    //消息在commitlog文件中的真实位移
                    long offset = byteBuffer.getLong();
                    //消息大小
                    int msgSize = byteBuffer.getInt();
                    //临时字段（暂时不用）
                    long tagsCode = byteBuffer.getLong();
                    if (offset >= 0 && msgSize > 0) {
//                        System.out.println("第" + i / 20 + "条消息在commitlog文件中的真实位置：" + offset);
                        //当前逻辑队列中，最后一条消息在commitlog文件中的真实位移
                        this.maxPhysicOffset = offset;
                        /**
                         * 1. 0+20，表示存储第一条消息后的可写入位置
                         * 2. 1+20，表示存储第二条消息后的可写入位置
                         * 3. ......
                         */
                        mappedFileOffset = i + UNIT_SIZE;
                    } else {
//                        System.out.println("consumeQueue队列文件异常！" + offset + "=" + msgSize);
                        break;
                    }
                }

                //确保每个文件的大小是一致的
                if (mappedFileOffset == mappFileSizeLogics) {
                    //下一个文件所在下标
                    index++;
                    if (index >= size) {
                        break;
                    } else {
                        mappFile = mappedFiles.get(index);
                        byteBuffer = mappFile.sliceByteBuffer();
                        //重置起始位移
                        processOffset = (int) mappFile.getFileFromOffset();
                        //重置该文件当前写入位移（每次遍历完一个文件后，都要重置该参数）
                        mappedFileOffset = 0;
                    }
                } else {
//                    System.out.println("consumequeue队列文件异常！");
                    break;
                }
            }
            //可写入位置
            processOffset += mappedFileOffset;
//            System.out.println(processOffset + "====>");
            //设置刷盘位置
            this.mapedFileQueue.setCommittedWhere(processOffset);
            this.mapedFileQueue.setFlushedWhere(processOffset);
            //处理数据文件的可写入位置
            this.mapedFileQueue.truncateDirtyFiles(processOffset);
        }
    }

    @Override
    public void dispatch(DispatchRequest dispatchRequest) {
        putMessage(dispatchRequest);
    }

    //保存
    public void putMessage(DispatchRequest dispatchRequest) {
        if (dispatchRequest != null) {
            boolean ok = false;
            for (int i = 0; i <= 30 && !ok; i++) {
                try {
                    ok = putMessage(dispatchRequest.getCommitLogOffset(), dispatchRequest.getMsgSize(), dispatchRequest.getConsumeQueueOffset());
                } catch (MyException e) {
                    System.out.println(dispatchRequest.getTopic() + "写入错误" + "，第几个消息：" + dispatchRequest.getConsumeQueueOffset());
                } catch (Exception e) {
                    System.out.println(dispatchRequest.getTopic() + "写入错误" + dispatchRequest.getCommitLogOffset() + "，第几个消息：" + dispatchRequest.getConsumeQueueOffset());
                }
                if (!ok) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {

                    }
                } else {
                    return;
                }
            }

        }
    }

    /**
     * @param offset
     * @param size
     * @param queueOffset -- 在queue队列中的第几个消息
     * @return
     */
    public boolean putMessage(long offset, int size, long queueOffset) {
        if (offset <= this.maxPhysicOffset) {
            return true;
        }
        try {
            this.byteBuffer.flip();
            this.byteBuffer.limit(UNIT_SIZE);
            this.byteBuffer.putLong(offset);
            this.byteBuffer.putInt(size);
            this.byteBuffer.putLong(0);//该字段暂时存放0
            //该记录应该在queue文件中的位置
            long logicoffset = queueOffset * UNIT_SIZE;
            MappFile mappFile = this.mapedFileQueue.getLastMappedFile(logicoffset, true);
            if (mappFile != null) {
                /**
                 * 1. queue文件中当前的位置（系统每次重启后，都会从下标0开始检测commitlog文件，如果consumequeue逻辑文件不存在，那么便恢复逻辑文件
                 * 2. consumequeue逻辑文件的位移必须与commit文件位移一致
                 */
                long currentOffset = mappFile.getFileFromOffset() + mappFile.getWrotePosition().get();

//                System.out.println("第几条消息：" + queueOffset + "，应当存储的位置：" + logicoffset + "，当前queue文件的位置：" + currentOffset + "-->" + mappFile.getPath());
                if (logicoffset != currentOffset) {
                    throw new MyException("写入逻辑队列错误，第" + queueOffset + "条消息应当存储的位置：" + logicoffset + "，当前queue文件可写入的位置：" + currentOffset + "，位置：" + mappFile.getPath());
                }
                boolean result = mappFile.appendMsg(this.byteBuffer.array());
                System.out.println("第" + queueOffset + "条消息存储:" + result + ",存储后的位置：" + mappFile.getWrotePosition().get() + "--" + mappFile.getPath());
                return result;
            }
        } catch (Exception e) {
            e.printStackTrace();
            throw new MyException("写入失败");
        }
        return false;
    }

    /**
     * 刷盘
     *
     * @param page
     */
    public void flush(int page) {
        //找需要刷盘的文件
        mapedFileQueue.flush(page);
    }

    public SelectMappedBufferResult getIndexBuffer(int index) {
        /**
         * 1. 一个消息大小 20K
         * 2. 第一个消息的位移： 0*20=0
         * 3. 第二个消息的位移： 1*20=20
         * 4. 第三个消息的位移：2*20=40
         */
        long offset = index * UNIT_SIZE;
        MappFile mappFile = this.mapedFileQueue.findMappedFileByOffset(offset, true);
        if (mappFile != null) {
//            System.out.println("不空");
            //公式：offset % fileSize  计算在第几个queue文件中（每个queue文件大小有限制）
            return mappFile.selectMappedBuffer((int) offset % fileSize);
        } else {
//            System.out.println("空的");
        }
        return null;
    }

    public long getConsumeQueueOffset() {
        long maxOffset = this.mapedFileQueue.getMaxOffset();
        //计算依据存储了多少个消息
        return maxOffset / UNIT_SIZE;
    }

    public int getFileSize() {
        return fileSize;
    }

    public String getTopic() {
        return topic;
    }

    public int getQueueId() {
        return queueId;
    }

    public long getMaxPhysicOffset() {
        return maxPhysicOffset;
    }

    public long getMinLogicOffset() {
        return minLogicOffset;
    }

    public static int getUnitSize() {
        return UNIT_SIZE;
    }

}
